package com.cgy.utils.zookeeper;

import com.cgy.utils.zookeeper.entity.NodeInfo;
import com.google.common.base.Charsets;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import lombok.Getter;
import lombok.Setter;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.BackgroundCallback;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.ProtectACLCreateModePathAndBytesable;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import static org.apache.curator.framework.imps.CuratorFrameworkState.STARTED;

/**
 * zookeeper工具类
 * @author CH
 *
 */
@Slf4j
@Accessors(fluent = true)
public class Zookeepers {
	/**
	 * zk地址
	 */
	private String ips;
	/**
	 * session超时
	 */
	private int sessionTimeoutMs = 30000;
	/**
	 * 路径超时
	 */
	private int connectionTimeoutMs = 30000;

	private ConcurrentHashMap<String, NodeInfo> nodeOldCache = new ConcurrentHashMap<>();
	private ConcurrentHashMap<String, NodeInfo> nodeCache = new ConcurrentHashMap<>();
	private CountDownLatch countDownLatch = new CountDownLatch(1);
	@Getter
	private CuratorFramework curatorFramework;
	private String root;
	
	private static final String EXTENSION_COMMA = ",";
	
	private PathChildrenCache pathChildrenCache;

	public Zookeepers(String ips) {
		this.ips = ips;
	}

	public Zookeepers(String ips, int connectionTimeoutMs) {
		this.ips = ips;
		this.connectionTimeoutMs = connectionTimeoutMs;
	}

	public Zookeepers(String ips, int sessionTimeoutMs, int connectionTimeoutMs) {
		this.ips = ips;
		this.sessionTimeoutMs = sessionTimeoutMs;
		this.connectionTimeoutMs = connectionTimeoutMs;
	}


	public static ZookeeperBuilder builder() {
		return new ZookeeperBuilder();
	}
	
	/**
	 * 设置节点
	 * @param name
	 * @param nodeInfo
	 */
	public final void nodeInfo(String name, NodeInfo nodeInfo) {
		nodeCache.put(name, nodeInfo);
	}
	/**
	 * 设置节点
	 * @param name
	 * @param nodeInfo
	 */
	public final void nodeUpdateInfo(String name, NodeInfo nodeInfo) {
		//备份
		NodeInfo nodeInfo2 = nodeCache.get(name);
		nodeOldCache.put(name, nodeInfo2);
		//
		nodeCache.put(name, nodeInfo);
	}
	/**
	 * 获取备份节点
	 * @param name
	 * @return 
	 */
	public final NodeInfo oldNodeInfo(String name) {
		return nodeOldCache.get(name);
	}
	/**
	 * 删除节点
	 * @param name
	 */
	public final void removeNodeInfo(String name) {
		nodeCache.remove(name);
	}

	/**
	 * 连接
	 */
	public final void connect() throws InterruptedException {
		if(StringUtils.isNotEmpty(ips)) {
			//重复策略
			RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
			this.curatorFramework = CuratorFrameworkFactory.newClient(ips, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
			this.curatorFramework.start();
            this.curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState) {
                	log.info("等待连接");
                   	if(newState.isConnected()) {
                       log.info("连接成功...");
                       countDownLatch.countDown();
                   }
                }
            });
			countDownLatch.await();
		} else {
			throw new IllegalArgumentException("ips must have a value");
		}
	}

	/**
	 * 释放连接
	 */
	public final void release() {
		if(isStart()) {
			this.curatorFramework.close();
			try {
				this.pathChildrenCache.close();
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 节点数据更新
	 * @param node 节点
	 * @return
	 * @throws Exception
	 */
	public final boolean updateData(final String node) throws Exception {
		return updateData(node, null, EXTENSION_COMMA, true, true);
	}
	/**
	 * 节点数据更新
	 * @param node 节点
	 * @param data 数据
	 * @return
	 * @throws Exception
	 */
	public final boolean updateData(final String node, final String data) throws Exception {
		return updateData(node, data, EXTENSION_COMMA, true, true);
	}

	/**
	 * 节点数据更新
	 * @param node 节点
	 * @param data 数据
	 * @param append 是否追加
	 * @return
	 * @throws Exception
	 */
	public final boolean updateData(final String node, final String data, boolean append) throws Exception {
		return updateData(node, data, EXTENSION_COMMA, append, true);
	}

	/**
	 * 节点数据更新
	 * @param node 节点
	 * @param data 数据
	 * @param append 是否追加
	 * @param duplicates 是否去重
	 * @return
	 * @throws Exception
	 */
	public final boolean updateData(final String node, final String data, String separator, boolean append, boolean duplicates) throws Exception {
		if(isStart() && exist(node)) {
			String data1 = getData(node);
			String newData = data;
			if(null != data1 && !"".equals(data1)) {
				if(append) {
					newData = data1 + separator + data;
				}
			}

			if(StringUtils.isBlank(separator)) {
				separator = EXTENSION_COMMA;
			}
			
			if(duplicates) {
				List<String> strings = Splitter.on(separator).splitToList(newData);
				Set<String> sets = new HashSet<>();
				sets.addAll(strings);
				newData = Joiner.on(separator).skipNulls().join(sets);
			}

			setData(node, newData);
		}
		return false;
	}
	/**
	 * 覆盖
	 * @param node
	 * @return
	 * @throws Exception 
	 */
	public final boolean override(final String node) throws Exception {
		return updateData(node, null, null, false, false);
	}
	
	/**
	 * 覆盖
	 * @param node
	 * @param data
	 * @return
	 * @throws Exception 
	 */
	public final boolean override(final String node, final String data) throws Exception {
		return updateData(node, data, null, false, false);
	}
	
	/**
	 * 覆盖
	 * @param node
	 * @param data
	 * @param separator
	 * @return
	 * @throws Exception 
	 */
	public final boolean override(final String node, final String data, final String separator) throws Exception {
		return updateData(node, data, separator, false, false);
	}
	/**
	 * 覆盖
	 * @param node
	 * @param data
	 * @param separator
	 * @return
	 * @throws Exception 
	 */
	public final boolean override(final String node, final String data, final String separator, final boolean duplicates) throws Exception {
		return updateData(node, data, separator, false, duplicates);
	}
	/**
	 * 监听
	 * @param node
	 * @throws Exception
	 */
	public final void monitor(final String node) throws Exception {
		if(isStart() && exist(node)) {
			this.pathChildrenCache = new PathChildrenCache(this.curatorFramework, node, false);
			try {
				pathChildrenCache.start();
				pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
					@Override
					public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
						PathChildrenCacheEvent.Type type = event.getType();

						final ChildData childData = event.getData();
						if(null != childData) {
							Stat childDataStat = childData.getStat();

							NodeInfo nodeInfo = new NodeInfo();
							nodeInfo.setData(childData.getPath())
									.setAversion(childDataStat.getAversion())
									.setCtime(childDataStat.getCtime())
									.setCversion(childDataStat.getCversion())
									.setCzxid(childDataStat.getCzxid())
									.setEphemeralOwner(childDataStat.getEphemeralOwner())
									.setMtime(childDataStat.getMtime())
									.setMzxid(childDataStat.getMzxid())
									.setNumChildren(childDataStat.getNumChildren())
									.setPzxid(childDataStat.getPzxid())
									.setVersion(childDataStat.getVersion())
									.setDataList(asList(childData.getData(), EXTENSION_COMMA))
									.setData(asString(childData.getData()))
									.setStatus(type);

							switch (type) {
								case CHILD_ADDED:
									log.info("节点添加{}", event.getData());
									nodeCache.put(childData.getPath(), nodeInfo);
									break;
								case CHILD_REMOVED:
									log.info("节点删除{}", event.getData());
									nodeCache.remove(childData.getPath());
									break;
								case CHILD_UPDATED:
									log.info("节点更新{}", event.getData());
									nodeCache.put(childData.getPath(), nodeInfo);
									break;
								default:
									break;
							}
						}
					}

					private List<String> asList(byte[] data, String extensionComma) {
						String s = asString(data);
						return Splitter.on(extensionComma).splitToList(s);
					}

					private String asString(byte[] data) {
						if(null != data) {
							return new String(data, Charsets.UTF_8);
						}
						return "";
					}
				});
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	/**
	 * 监听
	 * @param node
	 * @param listenable
	 * @throws Exception
	 */
	public final void monitor(final String node, final PathChildrenCacheListener listenable) throws Exception {
		if(isStart() && exist(node)) {
			pathChildrenCache = new PathChildrenCache(this.curatorFramework, node, false);
			try {
				pathChildrenCache.start();
				pathChildrenCache.getListenable().addListener(listenable);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 获取节点
	 * @return
	 */
	public final ConcurrentHashMap<String, NodeInfo> getCacheNode() {
		return nodeCache;
	}
	/**
	 * 获取子节点
	 * @param node
	 * @return
	 * @throws Exception
	 */
	public final List<String> children(final String node) throws Exception {
		if(isStart() && exist(node)) {
			List<String> strings = this.curatorFramework.getChildren().forPath(node);
			return strings;
		}
		return null;
	}
	/**
	 * 获取节点数据
	 * @param node 节点
	 * @return
	 * @throws Exception
	 */
	public final List<String> getDataList(final String node) throws Exception {
		return getData(node, EXTENSION_COMMA);
	}
	/**
	 * 获取节点数据
	 * @param node 节点
	 * @return
	 * @throws Exception
	 */
	public final String getData(final String node) throws Exception {
		List<String> dataList = getData(node, null);
		return null != dataList && !dataList.isEmpty() ? dataList.get(0) : null;
	}
	/**
	 * 获取节点数据
	 * @param node 节点
	 * @param separator 分隔符
	 * @return
	 * @throws Exception
	 */
	public final List<String> getData(final String node, final String separator) throws Exception {
		if(isStart() && exist(node)) {
			byte[] bytes = this.curatorFramework.getData().storingStatIn(new Stat()).forPath(node);
			final String dates = new String(bytes, Charsets.UTF_8);
			if(StringUtils.isNotEmpty(separator)) {
				return Splitter.on(separator).splitToList(dates);
			}
			return Arrays.asList(dates);
		}
		return Collections.emptyList();
	}
	/**
	 * 赋值
	 * @param node
	 * @param data
	 * @return
	 * @throws Exception
	 */
	public final String setData(final String node, final String data) throws Exception {
		if(isStart() && exist(node)) {
			byte[] bytes = null;
			if(null == data) {
				bytes = new byte[0];
			} else {
				bytes = data.getBytes(Charsets.UTF_8);
			}
			this.curatorFramework.setData().forPath(node, bytes);
			return data;
		}
		return null;
	}

	/**
	 * 赋值
	 * @param node 节点
	 * @param data 数据
	 * @return
	 * @throws Exception
	 */
	public final String setData(final String node, final List<String> data) throws Exception {
		return setData(node, data, EXTENSION_COMMA);
	}

	/**
	 * 赋值
	 * @param node 节点
	 * @param data 数据
	 * @param separator 分隔符
	 * @return
	 * @throws Exception
	 */
	public final String setData(final String node, final List<String> data, final String separator) throws Exception {
		String newData = null;
		if(StringUtils.isNotEmpty(separator)) {
			newData = Joiner.on(separator).join(data);
		}
		return setData(node, newData);
	}
	/**
	 * 删除节点
	 * @param node
	 * @return
	 * @throws Exception
	 */
	public final boolean delete(final String node) throws Exception {
		if(isStart()) {
			//guaranteed 保障机制，若未删除成功，只要会话有效会在后台一直尝试删除
			Void aVoid = this.curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(node);
			return null == aVoid;
		}
		return true;
	}

	/**
	 * 是否存在
	 * @param node
	 * @return
	 * @throws Exception
	 */
	public final boolean exist(final String node) throws Exception {
		if(isStart()) {
			Stat stat = this.curatorFramework.checkExists().forPath(node);
			return null != stat;
		}
		return false;
	}
	/**
	 * 创建节点(默认,)
	 * @param node 节点
	 * @return
	 * @throws Exception
	 */
	public final String create(final String node) throws Exception {
		return create(node, null, CreateMode.EPHEMERAL);
	}
	/**
	 * 创建节点(默认,)
	 * @param node 节点
	 * @return
	 * @throws Exception
	 */
	public final String create(final String node, final String data) throws Exception {
		return create(node, data, CreateMode.EPHEMERAL);
	}
    /**
     * 创建节点(默认,)
     * @param node 节点
     * @param createMode 类型
     * @return
     * @throws Exception
     */
    public final String create(final String node, final CreateMode createMode) throws Exception {
        return create(node, null, createMode);
    }
    /**
     * 创建节点(默认,)
     * @param node 节点
     * @param createMode 类型
     * @return
     * @throws Exception
     */
    public final String root(final String node, final CreateMode createMode) throws Exception {
    	this.root = node;
    	return create(node, null, createMode);
    }
	/**
	 * 创建节点(默认,)
	 * @param node 节点
	 * @param data 数据
	 * @return
	 * @throws Exception
	 */
	public final String create(final String node, final List<String> data) throws Exception {
		return create(node, data, EXTENSION_COMMA);
	}
	/**
	 * 创建节点
	 * @param node 节点
	 * @param data 数据
	 * @param separator 分隔符
	 * @return
	 * @throws Exception
	 */
	public final String create(final String node, final List<String> data, final String separator) throws Exception {
		String newData = null;
		if(null != data && !data.isEmpty()) {
			newData = Joiner.on(separator).join(data);
		}
		return create(node, newData, CreateMode.EPHEMERAL);
	}
	/**
	 * 创建节点
	 * @param node 节点
	 * @param data 数据
	 * @param createMode 节点类型
	 * <p>CreateMode.EPHEMERAL</p>
	 * <p>CreateMode.PERSISTENT</p>
	 * @return
	 * @throws Exception
	 */
	public final String create(final String node, final String data, CreateMode createMode) throws Exception {
		if(isStart()) {
			if(!exist(node)) {
				ProtectACLCreateModePathAndBytesable<String> stringProtectACLCreateModePathAndBytesable = this.curatorFramework.create().creatingParentsIfNeeded();
				stringProtectACLCreateModePathAndBytesable.withMode(createMode);
				if(StringUtils.isNotEmpty(data)) {
					stringProtectACLCreateModePathAndBytesable.inBackground(new BackgroundCallback() {
						
						@Override
						public void processResult(CuratorFramework arg0, CuratorEvent curatorEvent) throws Exception {
							log.info("注册信息:node:{}, status:{}, data:{} ", node, curatorEvent.getStat(), data);
						}
					}).forPath(node, data.getBytes(Charsets.UTF_8));
					return node;
				} else {
					stringProtectACLCreateModePathAndBytesable.inBackground(new BackgroundCallback() {
						@Override
						public void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {
							log.info("注册信息:node:{}, status:{} ", node, curatorEvent.getStat());
						}
					}).forPath(node, new byte[0]);
					return node;
				}
			} else {
				log.warn("注册信息:Node named {} already exists", node);
			}
		}
		return null;
	}

	/**
	 * 正常启动状态
	 * @return
	 */
	public final boolean isStart() {
		CuratorFrameworkState state = this.curatorFramework.getState();
		return null != this.curatorFramework && state == STARTED;
	}


	@Getter
	@Setter
	@Accessors(fluent = true)
	public static final class ZookeeperBuilder {
		private String ips;
		private int sessionTimeoutMs = 30000;
		private int connectionTimeoutMs = 30000;

		public Zookeepers build() {
			return new Zookeepers(ips, sessionTimeoutMs, connectionTimeoutMs);
		}

	}
	
	public final String root() {
		return this.root;
	}


}
